package com.gandong8.trace.tool.consumer.kafka;

import com.gandong8.trace.tool.util.TraceUtil;
import com.gandong8.trace.tool.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;

/**
 * MessageListener
 * AcknowledgingMessageListener
 * @param <T>
 */
@Slf4j
public abstract class AbstractKafkaListener<T> implements AcknowledgingMessageListener<String,String> {

    private Class<T> recordClazz;



    public AbstractKafkaListener() {
        parseClazz();
    }


    @SuppressWarnings("unchecked")
    @Override
    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        for (Header header : record.headers()) {
            if (TraceUtil.TRACE_TAG.equals(header.key())) {
                TraceUtil.setLogTrace(new String(header.value(), StandardCharsets.UTF_8));
                break;
            }
        }
        String text = record.value();
        log.info("收到消息:topic={},offset={},partition={},消息内容={}", record.topic(), record.offset(), record.partition(), text);
        acknowledgment.acknowledge();
        try {
            if ("String".equals(recordClazz.getSimpleName())) {
                execute((T) text);
            } else {
                T obj = JsonUtil.toObj(text, recordClazz);
                execute(obj);
            }
        } catch (Exception e) {
            log.error("消息处理异常", e);
        }

    }

    protected abstract void execute(T message);



    @SuppressWarnings("unchecked")
    private void parseClazz() {
        Type type = getClass().getGenericSuperclass();
        if (type instanceof ParameterizedType) {
            recordClazz = (Class<T>) ((ParameterizedType) type).getActualTypeArguments()[0];
        }
    }




}
